package com.cbh.binlog.core;

import com.cbh.binlog.common.LRU;
import com.cbh.binlog.common.SpringUtil;
import com.cbh.binlog.config.BinlogClientConfig;
import com.cbh.binlog.entity.BinaryLogEventCallback;
import com.cbh.binlog.entity.Column;
import com.cbh.binlog.link.BinaryLogEventListener;
import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData;
import com.github.shyiko.mysql.binlog.event.QueryEventData;
import com.github.shyiko.mysql.binlog.event.UpdateRowsEventData;
import com.github.shyiko.mysql.binlog.event.WriteRowsEventData;
import lombok.extern.slf4j.Slf4j;
import net.sf.jsqlparser.parser.CCJSqlParserUtil;
import net.sf.jsqlparser.schema.Table;
import net.sf.jsqlparser.statement.Statement;
import net.sf.jsqlparser.statement.alter.Alter;
import net.sf.jsqlparser.statement.alter.RenameTableStatement;
import net.sf.jsqlparser.statement.create.table.ColumnDefinition;
import net.sf.jsqlparser.statement.create.table.CreateTable;
import net.sf.jsqlparser.statement.drop.Drop;
import org.springframework.lang.Nullable;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import javax.sql.DataSource;
import java.io.Serializable;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * Binlog信息 核心解析器
 *
 * @author ChenBiHui
 * @date 2025/1/16
 */
@Slf4j
@SuppressWarnings("all")
public final class SchemaParser {

    /**
     * 事务
     */
    private static final String BEGIN = "begin";

    /**
     * 匹配位
     */
    private static final int INDEX = 4;

    /**
     * 建库语句
     */
    private static final String CREATE_SCHEMA = "(?i)CREATE\\s+(DATABASE|SCHEMA)\\s+(IF\\s+NOT\\s+EXISTS\\s+)?(`)?(\\w+)(`)?";

    /**
     * 建库语句 - 正则
     */
    private static final Pattern CREATE_SCHEMA_PATTERN = Pattern.compile(CREATE_SCHEMA);

    /**
     * 删库语句
     */
    private static final String DROP_SCHEMA = "(?i)DROP\\s+(DATABASE|SCHEMA)\\s+(IF\\s+EXISTS\\s+)?(`)?(\\w+)(`)?";

    /**
     * 删库语句 - 正则
     */
    private static final Pattern DROP_SCHEMA_PATTERN = Pattern.compile(DROP_SCHEMA);

    /**
     * 表ID与归属库关系
     */
    private static final Map<Long, String> ID_WITH_SCHEMA = new LRU<>(222);

    /**
     * 表ID与表名关系
     */
    private static final Map<Long, String> ID_WITH_TABLE = new LRU<>(222);

    /**
     * 库名与(表名-列清单)关系
     */
    private static final Map<String, Map<String, List<Column>>> SCHEMA = new HashMap<>();

    /**
     * 库.表与通道线程绑定关系
     */
    private static final Map<String, ThreadPoolTaskExecutor> BINDING = new HashMap<>();

    /**
     * 单例单连接连接池
     */
    private static final DataSourceSchema DATA_SOURCE_SCHEMA = new DataSourceSchema();

    /**
     * 获取连接池
     *
     * @return 数据库连接池
     */
    static DataSource dataSource() {
        try {
            return DATA_SOURCE_SCHEMA.dataSource();
        } catch (ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
    }


    /**
     * 关系绑定
     *
     * @param payload 库名.表名
     * @param channel 通道
     */
    static void binding(String payload, ThreadPoolTaskExecutor channel) {
        BINDING.putIfAbsent(payload, channel);
    }

    /**
     * 关系解绑
     *
     * @param payload 库名.表名
     */
    static ThreadPoolTaskExecutor unbinding(String payload) {
        return BINDING.remove(payload);
    }

    /**
     * 获取通道
     *
     * @param schema 库名
     * @param table  表名
     * @return 通道
     */
    static ThreadPoolTaskExecutor channel(String schema, String table) {
        return BINDING.getOrDefault(compose(schema, table), Entrance.defaulted());
    }

    /**
     * 启动加载表结构
     *
     * @param schema 库名
     * @param table  表名
     * @param column 列对象
     */
    static void save(String schema, String table, Column column) {
        SCHEMA.computeIfAbsent(schema, k -> new HashMap<>())
                .computeIfAbsent(table, k -> new ArrayList<>())
                .add(column);
    }

    /**
     * 启动加载表结构
     *
     * @param schema  库名
     * @param table   表名
     * @param columns 列对象
     */
    static void save(String schema, String table, List<Column> columns) {
        List<Column> isLatest = SCHEMA.computeIfAbsent(schema, k -> new HashMap<>())
                .computeIfAbsent(table, k -> new ArrayList<>());
        if (isLatest.isEmpty()) {
            isLatest.addAll(columns);
        }
    }

    /**
     * binlog监听器加载关系映射
     *
     * @param schema  库名
     * @param table   表名
     * @param tableId 表ID
     */
    static void load(String schema, String table, Long tableId) {
        ID_WITH_SCHEMA.put(tableId, schema);
        ID_WITH_TABLE.put(tableId, table);
    }

    /**
     * 解析调整事件
     *
     * @param query 调整事件
     */
    static void parse(QueryEventData query) {
        String sql = query.getSql();
        if (sql == null) {
            return;
        }
        if (BEGIN.equalsIgnoreCase(sql)) {
            return;
        }
        try {
            Matcher isSchemaOperate = CREATE_SCHEMA_PATTERN.matcher(sql);
            if (isSchemaOperate.find()) {
                String schema = isSchemaOperate.group(INDEX);
                log.info("parse query event with [create-scheme] the schema name is : {}", schema);
                return;
            }
            isSchemaOperate = DROP_SCHEMA_PATTERN.matcher(sql);
            if (isSchemaOperate.find()) {
                String schema = isSchemaOperate.group(INDEX);
                log.info("parse query event with [drop-scheme] the schema name is : {}", schema);
                Map<String, List<Column>> remove = SCHEMA.remove(schema);
                if (remove == null) {
                    return;
                }
                for (String table : remove.keySet()) {
                    BINDING.remove(compose(schema, table));
                }
                return;
            }
            String schema = SchemaParser.modify(SchemaParser.hasText(query.getDatabase()) ? query.getDatabase() : null);
            String table;
            Statement parse = CCJSqlParserUtil.parse(sql);
            if (parse instanceof RenameTableStatement) {
                RenameTableStatement trans = (RenameTableStatement) parse;
                Set<Map.Entry<Table, Table>> targets = trans.getTableNames();
                for (Map.Entry<Table, Table> next : targets) {
                    Table old = next.getKey();
                    Table latest = next.getValue();
                    if (old == null || latest == null) {
                        continue;
                    }
                    if (schema == null) {
                        schema = SchemaParser.modify(SchemaParser.hasText(old.getSchemaName()) ? old.getSchemaName() : SchemaParser.hasText(latest.getSchemaName()) ? latest.getSchemaName() : null);
                    }
                    table = SchemaParser.modify(SchemaParser.hasText(old.getName()) ? old.getName() : null);
                    if (schema == null || table == null) {
                        continue;
                    }
                    ThreadPoolTaskExecutor release = SchemaParser.unbinding(SchemaParser.compose(schema, table));
                    List<Column> transfer = null;
                    Map<String, List<Column>> map = SCHEMA.get(schema);
                    if (map != null) {
                        transfer = map.remove(table);
                    }
                    if (transfer == null || transfer.isEmpty()) {
                        continue;
                    }
                    table = SchemaParser.modify(SchemaParser.hasText(latest.getName()) ? latest.getName() : null);
                    if (table == null) {
                        continue;
                    }
                    map.put(table, transfer);
                    SchemaParser.binding(SchemaParser.compose(schema, table), release);
                }
                return;
            }
            if (parse instanceof CreateTable) {
                CreateTable trans = (CreateTable) parse;
                Table move = trans.getTable();
                if (move == null) {
                    return;
                }
                if (schema == null) {
                    schema = SchemaParser.modify(SchemaParser.hasText(move.getSchemaName()) ? move.getSchemaName() : null);
                }
                table = SchemaParser.modify(SchemaParser.hasText(move.getName()) ? move.getName() : null);
                if (schema == null || table == null) {
                    return;
                }
                BinlogClientConfig config = SpringUtil.getBean(BinlogClientConfig.class);
                SchemaParser.refresh(schema, table, trans);
                SchemaParser.binding(SchemaParser.compose(schema, table), Entrance.match(config.getLoad()));
                return;
            }
            if (parse instanceof Drop) {
                Drop trans = (Drop) parse;
                Table move = trans.getName();
                if (move == null) {
                    return;
                }
                if (schema == null) {
                    schema = SchemaParser.modify(SchemaParser.hasText(move.getSchemaName()) ? move.getSchemaName() : null);
                }
                table = SchemaParser.modify(SchemaParser.hasText(move.getName()) ? move.getName() : null);
                if (schema == null || table == null) {
                    return;
                }
                Map<String, List<Column>> map = SCHEMA.get(schema);
                if (map != null) {
                    map.remove(table);
                }
                SchemaParser.unbinding(SchemaParser.compose(schema, table));
                return;
            }
            if (parse instanceof Alter) {
                Alter trans = (Alter) parse;
                Table move = trans.getTable();
                if (move == null) {
                    return;
                }
                if (schema == null) {
                    schema = SchemaParser.modify(SchemaParser.hasText(move.getSchemaName()) ? move.getSchemaName() : null);
                }
                table = SchemaParser.modify(SchemaParser.hasText(move.getName()) ? move.getName() : null);
                if (schema == null || table == null) {
                    return;
                }
                Map<String, List<Column>> map = SCHEMA.get(schema);
                if (map != null) {
                    map.remove(table);
                }
                BinlogClientConfig config = SpringUtil.getBean(BinlogClientConfig.class);
                SchemaParser.refresh(config, schema, table);
            }
        } catch (Throwable e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 解析写入事件
     *
     * @param insert 写入事件
     * @return 列清单对象
     */
    static List<List<Column>> parse(WriteRowsEventData insert) {
        List<List<Column>> result = new ArrayList<>();
        List<Column> columns = SchemaParser.find(insert.getTableId());
        if (columns.isEmpty()) {
            return result;
        }
        List<Serializable[]> rows = insert.getRows();
        if (rows.isEmpty()) {
            return result;
        }
        for (Serializable[] row : rows) {
            List<Column> copy = copy(columns);
            for (int i = 0; i < row.length; i++) {
                Serializable value = row[i];
                copy.get(i).setAfterValue(value);
            }
            result.add(copy);
        }
        return result;
    }

    /**
     * 解析更新事件
     *
     * @param update 更新事件
     * @return 列清单对象
     */
    static List<List<Column>> parse(UpdateRowsEventData update) {
        List<List<Column>> result = new ArrayList<>();
        List<Column> columns = SchemaParser.find(update.getTableId());
        if (columns.isEmpty()) {
            return result;
        }
        List<Map.Entry<Serializable[], Serializable[]>> rows = update.getRows();
        if (rows.isEmpty()) {
            return result;
        }
        for (Map.Entry<Serializable[], Serializable[]> row : rows) {
            List<Column> copy = copy(columns);
            Serializable[] before = row.getKey();
            Serializable[] after = row.getValue();
            for (int i = 0; i < before.length; i++) {
                Serializable value = before[i];
                copy.get(i).setBeforeValue(value);
            }
            for (int i = 0; i < after.length; i++) {
                Serializable value = after[i];
                copy.get(i).setAfterValue(value);
            }
            result.add(copy);
        }
        return result;
    }

    /**
     * 解析删除事件
     *
     * @param delete 删除事件
     * @return 列清单对象
     */
    static List<List<Column>> parse(DeleteRowsEventData delete) {
        List<List<Column>> result = new ArrayList<>();
        List<Column> columns = SchemaParser.find(delete.getTableId());
        if (columns.isEmpty()) {
            return result;
        }
        List<Serializable[]> rows = delete.getRows();
        if (rows.isEmpty()) {
            return result;
        }
        for (Serializable[] row : rows) {
            List<Column> copy = copy(columns);
            for (int i = 0; i < row.length; i++) {
                Serializable value = row[i];
                copy.get(i).setBeforeValue(value);
            }
            result.add(copy);
        }
        return result;
    }

    /**
     * 构建回调
     *
     * @param schema 库名
     * @param table  表名
     * @param rows   数据行
     * @return 回调对象
     */
    static BinaryLogEventCallback callback(String schema, String table, List<List<Column>> rows) {
        BinaryLogEventCallback callback = new BinaryLogEventCallback();
        callback.setSchema(schema);
        callback.setTable(table);
        callback.setRows(rows);
        return callback;
    }

    /**
     * 根据表ID获取库名
     *
     * @param tableId 表ID
     * @return 库名
     */
    static String findSchema(Long tableId) {
        return ID_WITH_SCHEMA.getOrDefault(tableId, "");
    }

    /**
     * 根据表ID获取表名
     *
     * @param tableId 表ID
     * @return 表名
     */
    static String findTable(Long tableId) {
        return ID_WITH_TABLE.getOrDefault(tableId, "");
    }

    /**
     * 合并
     *
     * @param schema 库名
     * @param table  表名
     * @return 合并结果
     */
    public static String compose(String schema, String table) {
        return schema + BinaryLogEventListener.DOL + table;
    }

    /**
     * 字符串合法性判断
     *
     * @param str 目标字符串
     * @return 是否合法
     */
    public static boolean hasText(@Nullable String str) {
        return (str != null && !str.isEmpty() && containsText(str));
    }

    /**
     * 判断是否存在空格
     *
     * @param str 目标字符串
     * @return 是否合法
     */
    private static boolean containsText(CharSequence str) {
        int strLen = str.length();
        for (int i = 0; i < strLen; i++) {
            if (!Character.isWhitespace(str.charAt(i))) {
                return true;
            }
        }
        return false;
    }

    /**
     * 通过表ID,找到映射关系
     *
     * @param tableId 表ID
     * @return 列清单
     */
    private static List<Column> find(Long tableId) {
        String schema = ID_WITH_SCHEMA.get(tableId);
        if (schema == null) {
            log.warn("Current table-id {}, can't find any belong schema", tableId);
            return new ArrayList<>();
        }
        String table = ID_WITH_TABLE.get(tableId);
        if (table == null) {
            log.warn("Current table-id {}, can't find any belong table", tableId);
            return new ArrayList<>();
        }
        Map<String, List<Column>> map = SCHEMA.get(schema);
        if (map == null) {
            log.warn("Current schema {}, can't find any belong tables", schema);
            return new ArrayList<>();
        }
        List<Column> columns = map.get(table);
        if (columns == null) {
            log.warn("Current schema.table {}.{}, can't find any belong columns", schema, table);
            return new ArrayList<>();
        }
        return columns;
    }

    /**
     * 复制列对象清单
     *
     * @param columns 匹配列清单
     * @return 复制品
     */
    private static List<Column> copy(List<Column> columns) {
        return columns.stream()
                .map(item -> {
                    Column trans = new Column();
                    trans.setName(item.getName());
                    trans.setType(item.getType());
                    trans.setPosition(item.getPosition());
                    return trans;
                })
                .sorted(Comparator.comparingInt(Column::getPosition))
                .collect(Collectors.toList());
    }

    /**
     * 修正多余字符
     *
     * @param name 目标库名/表名
     * @return 修正值
     */
    private static String modify(String name) {
        if (name == null) {
            return null;
        }
        return name.replace("`", "").toLowerCase();
    }

    /**
     * 刷新缓存表结构关系映射
     *
     * @param schema 库名
     * @param table  表名
     * @param trans  解析语法树
     */
    private static void refresh(String schema, String table, CreateTable trans) {
        AtomicInteger index = new AtomicInteger();
        List<ColumnDefinition> columnDefinitions = trans.getColumnDefinitions();
        if (columnDefinitions.isEmpty()) {
            return;
        }
        SchemaParser.save(schema, table, columnDefinitions.stream()
                .map(columnDefinition -> {
                    Column column = new Column();
                    column.setName(columnDefinition.getColumnName());
                    column.setType(columnDefinition.getColDataType().getDataType());
                    column.setPosition(index.getAndIncrement());
                    return column;
                }).collect(Collectors.toList()));
    }

    /**
     * 刷新缓存表结构关系映射
     *
     * @param config 配置类
     * @param schema 库名
     * @param table  表名
     */
    private static void refresh(BinlogClientConfig config, String schema, String table) {
        Connection connection = null;
        PreparedStatement statement = null;
        ResultSet resultSet = null;
        try {
            connection = SchemaParser.dataSource().getConnection();
            statement = connection.prepareStatement("SELECT TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, DATA_TYPE, ORDINAL_POSITION " +
                    "FROM INFORMATION_SCHEMA.COLUMNS " +
                    "WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? " +
                    "ORDER BY TABLE_SCHEMA, TABLE_NAME, ORDINAL_POSITION");
            statement.setString(1, schema);
            statement.setString(2, table);
            resultSet = statement.executeQuery();
            while (resultSet.next()) {
                Column column = new Column();
                column.setName(resultSet.getString("COLUMN_NAME"));
                column.setType(resultSet.getString("DATA_TYPE"));
                column.setPosition(resultSet.getInt("ORDINAL_POSITION") - 1);
                SchemaParser.save(schema, table, column);
            }
        } catch (Throwable e) {
            log.error(e.toString(), e);
        } finally {
            if (resultSet != null) {
                try {
                    resultSet.close();
                } catch (Throwable ignore) {
                }
            }
            if (statement != null) {
                try {
                    statement.close();
                } catch (Throwable ignore) {
                }
            }
            if (connection != null) {
                try {
                    connection.close();
                } catch (Throwable ignore) {
                }
            }
        }
    }
}
